package com.minimalist.common.config.sse;

import jakarta.annotation.PreDestroy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.jetbrains.annotations.NotNull;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * SSE绠＄悊鏈嶅姟绫? * 璐熻矗绠＄悊SSE杩炴帴銆佹暟鎹帹閫佺瓑鍔熻兘
 *
 * @author 11987
 */
@Service
public class SseManagerService{
    private static final Logger log = LoggerFactory.getLogger(SseManagerService.class);


    /**
     * 数据更新间隔（秒）
     */
    public static final long DATA_UPDATE_INTERVAL = 30;

    /**
     * SSE连接超时时间（毫秒）
     */
    public static final long SSE_TIMEOUT = 5 * 60 * 1000;

    /**
     * 存储所有活跃的SSE连接
     */
    private final Map<String, SseEmitter> sseEmitters = new ConcurrentHashMap<>();

    /**
     * 定时任务执行器
     */
    private volatile ScheduledExecutorService scheduler = createScheduledThreadPool();

    /**
     * 数据推送任务Future
     */
    private volatile ScheduledFuture<?> pushTaskFuture;

    /**
     * 数据推送任务是否已启动
     */
    private volatile boolean pushTaskStarted = false;

    /**
     * 创建定时任务线程池
     *
     * @return 定时任务执行器
     */
    private ScheduledExecutorService createScheduledThreadPool(){
        return new ScheduledThreadPoolExecutor(2, new ThreadFactory(){
            private final AtomicInteger threadNumber = new AtomicInteger(1);

            @Override
            public Thread newThread(@NotNull Runnable r){
                Thread thread = new Thread(r, "SSE-Scheduler-"
                        + threadNumber.getAndIncrement());
                thread.setDaemon(true);
                return thread;
            }
        });
    }

    /**
     * 创建SSE连接
     *
     * @param clientId 客户端ID
     * @return SSE发射器
     */
    public SseEmitter createConnection(String clientId){
        // 生成客户端ID
        if (clientId == null || clientId.trim().isEmpty()){
            clientId = "client_" + System.currentTimeMillis() + "_" + UUID.randomUUID().toString().substring(0, 8);
        }

        // 确保clientId为final，以便在lambda表达式中使用
        final String finalClientId = clientId;

        log.info("SSE连接建立，客户端ID: {}", finalClientId);

        // 创建SSE发射器
        SseEmitter emitter = new SseEmitter(SSE_TIMEOUT);

        // 存储连接
        sseEmitters.put(finalClientId, emitter);

        // 设置连接完成回调
        emitter.onCompletion(() -> {
            log.info("SSE连接完成，客户端ID: {}", finalClientId);
            sseEmitters.remove(finalClientId);
        });

        // 设置连接超时回调
        emitter.onTimeout(() -> {
            log.info("SSE连接超时，客户端ID: {}", finalClientId);
            sseEmitters.remove(finalClientId);
        });

        // 设置连接错误回调
        emitter.onError(throwable -> {
            log.error("SSE连接错误，客户端ID: {}, 错误信息: {}", finalClientId, throwable.getMessage());
            sseEmitters.remove(finalClientId);
        });

        try{
            // 发送连接消息
            Map<String, Object> connectMsg = new HashMap<>(4);
            connectMsg.put("type", "connect");
            connectMsg.put("clientId", finalClientId);
            connectMsg.put("message", "SSE连接建立");
            connectMsg.put("timestamp", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));

            emitter.send(SseEmitter.event()
                    .name("connect")
                    .data(connectMsg));

        } catch (IOException e){
            log.error("发送连接消息失败，客户端ID: {}", finalClientId, e);
            sseEmitters.remove(finalClientId);
        }

        return emitter;
    }

    /**
     * 数据推送
     *
     * @param dataSupplier 数据供应者
     */
    public synchronized void startDataPush(DataSupplier dataSupplier){
        if (pushTaskStarted){
            log.warn("数据推送已启动");
            return;
        }

        // 妫€鏌ョ嚎绋嬫睜鏄惁宸插叧闂紝濡傛灉宸插叧闂垯閲嶆柊鍒涘缓
        if (scheduler.isShutdown() || scheduler.isTerminated()){
            log.info("绾跨▼姹犲凡鍏抽棴锛岄噸鏂板垱寤虹嚎绋嬫睜");
            scheduler = createScheduledThreadPool();
        }

        // 启动定时推送任务
        pushTaskFuture = scheduler.scheduleAtFixedRate(() -> broadcastData(dataSupplier.get()),
                DATA_UPDATE_INTERVAL, DATA_UPDATE_INTERVAL, TimeUnit.SECONDS);

        pushTaskStarted = true;
        log.info("数据推送已启动");
    }

    /**
     * 停止数据推送
     */
    public synchronized void stopDataPush(){
        if (pushTaskFuture != null && !pushTaskFuture.isCancelled()){
            pushTaskFuture.cancel(false);
            pushTaskFuture = null;
        }
        pushTaskStarted = false;
        log.info("数据推送已停止");
    }

    /**
     * 数据推送
     *
     * @param data 数据
     */
    public void pushDataNow(Map<String, Object> data){
        broadcastData(data);
    }

    /**
     * 数据推送
     *
     * @param clientId 客户端ID
     * @param data     数据
     */
    public void sendDataToClient(String clientId, Map<String, Object> data){
        SseEmitter emitter = sseEmitters.get(clientId);
        if (emitter == null){
            log.warn("客户端ID不存在", clientId);
            return;
        }

        try{
            emitter.send(SseEmitter.event()
                    .name("dashboard-data")
                    .data(data));
        } catch (IOException e){
            log.warn("发送数据失败", e.getMessage());
            sseEmitters.remove(clientId);
        }
    }

    /**
     * 广播数据
     *
     * @param data 数据
     */
    public void broadcastData(Map<String, Object> data){
        if (sseEmitters.isEmpty()){
            return;
        }

        log.debug("开始向{}个客户端推送数据", sseEmitters.size());

        // 向所有客户端发送数据
        List<String> disconnectedClients = new ArrayList<>();

        sseEmitters.forEach((clientId, emitter) -> {
            try{
                emitter.send(SseEmitter.event()
                        .name("dashboard-data")
                        .data(data));
            } catch (IOException e){
                log.warn("发送数据失败", e.getMessage());
                disconnectedClients.add(clientId);
            }
        });

        // 绉婚櫎鏂紑杩炴帴鐨勫鎴风
        disconnectedClients.forEach(sseEmitters::remove);
    }

    /**
     * 获取连接信息
     *
     * @return 连接信息
     */
    public Map<String, Object> getConnectionInfo(){
        Map<String, Object> result = new HashMap<>(4);
        result.put("connectionCount", sseEmitters.size());
        result.put("clientIds", new ArrayList<>(sseEmitters.keySet()));
        return result;
    }

    /**
     * 关闭所有连接
     */
    public void closeAllConnections(){
        sseEmitters.forEach((clientId, emitter) -> {
            try{
                emitter.complete();
            } catch (Exception e){
                log.warn("关闭连接失败", clientId, e.getMessage());
            }
        });
        sseEmitters.clear();
        log.info("已关闭所有SSE连接");
    }

    /**
     * 获取连接数量
     *
     * @return 连接数量
     */
    public int getConnectionCount(){
        return sseEmitters.size();
    }

    /**
     * 判断客户端是否已连接
     *
     * @param clientId 客户端ID
     * @return 是否已连接
     */
    public boolean isClientConnected(String clientId){
        return sseEmitters.containsKey(clientId);
    }

    /**
     * 搴旂敤鍏抽棴鏃剁殑娓呯悊宸ヤ綔
     */
    @PreDestroy
    public void destroy(){
        log.info("SseManagerService正在关闭...");

        // 停止数据推送任务
        stopDataPush();

        // 关闭所有SSE连接
        closeAllConnections();

        // 关闭线程池
        if (scheduler != null && !scheduler.isShutdown()){
            scheduler.shutdown();
            try{
                if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)){
                    scheduler.shutdownNow();
                }
            } catch (InterruptedException e){
                scheduler.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }

        log.info("SseManagerService已关闭");
    }

    /**
     * 数据供应
     */
    @FunctionalInterface
    public interface DataSupplier{
        /**
         * 数据供应
         *
         * @return 数据
         */
        Map<String, Object> get();
    }
}
